package top.v5it.logback.push.convergence;

import lombok.Getter;
import lombok.Setter;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 计数收敛策略，消息累计到达一定数量推送
 *
 * @author zhanpu
 * @date 2022/9/8
 */
@Setter
@Getter
public class CountConvergenceStrategy implements ConvergenceStrategy {

    private final static ConcurrentMap<String, AtomicLong> counter = new ConcurrentHashMap<>();

    /**
     * 限制大小
     */
    private long limit;

    /**
     * 按消息体中消息编号统计单条消息到达数量，当单条消息数量达到{@link CountConvergenceStrategy#limit}限制就推送此条消息并删除计数器
     *
     * @param payload 消息体
     * @return
     */
    @Override
    public boolean isPush(byte[] payload) {
        return getMsgId(payload).map(t -> {
            if (counter.putIfAbsent(t, new AtomicLong(1)) == null) return false;
            final AtomicLong value = counter.computeIfPresent(t, (k, v) -> {
                v.incrementAndGet();
                return v.compareAndSet(limit, 0) ? null : v;
            });
            return Objects.isNull(value);
        }).orElse(Boolean.FALSE);
    }
}
